package com.we.risk.attribution.correct;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.we.utils.CommonTimeTools;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.text.SimpleDateFormat;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Supplier;

import static com.we.risk.attribution.correct.Common.*;
import static com.we.risk.attribution.AttributionInfo.*;
import static com.we.risk.attribution.correct.CmsRuleInfo.getCmsRuleInfo;
import static com.we.risk.attribution.correct.Common.DEVICEID;


public class DorisAsyncRichMapFunction extends RichAsyncFunction<UserInfo, JSONObject> {
    public static Logger LOG = LoggerFactory.getLogger(DorisAsyncRichMapFunction.class);

    public static CmsRuleInfo ruleInfo;
    public static SimpleDateFormat sdf;
    public static final String SPLITKEY = "@we@";

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        CmsRuleInfo tmpCmsRuleInfo = getCmsRuleInfo();
        if (tmpCmsRuleInfo != null) {
            ruleInfo = tmpCmsRuleInfo;
        } else if (ruleInfo == null) {
            throw new Exception("Init cmsRuleInfo Excep!!");
        }
        sdf = new SimpleDateFormat("yyyy-MM-dd");
    }

    @Override
    public void asyncInvoke(UserInfo user, ResultFuture<JSONObject> resultFuture)
            throws Exception {
        // 判断是否修正
        LOG.info("## asyncInvoke UserInfo: " + user.toString());
        JSONObject resJson = correctAttributionProcess(user);
        LOG.info("resJson: " + resJson.toString());
        final String deviceId = user.getDeviceId();
        final long uid = user.getUid();
        final String action = user.getAction();

        Future<JSONObject> droisResult =
                DruidUtils.executorService.submit(
                        new Callable<JSONObject>() {
                            @Override
                            public JSONObject call() throws Exception {

                                PreparedStatement statement = null;
                                ResultSet resultSet = null;
                                Connection connection = null;
                                try {
                                    LOG.info("##Druid input resJson: " + resJson.toString());
//                                    resJson.put(CORRECTENDTIME, System.currentTimeMillis());
                                    JSONArray attributionRes = resJson.getJSONArray(ATTRBURES);
                                    if(attributionRes != null) {
                                        int size = attributionRes.size();
                                        LOG.info("attributionRes size: " + size);
                                        for (int i = 0; i < size; i++) {
                                            JSONObject bucketJson = attributionRes.getJSONObject(i);
                                            LOG.info("bucketJson: " + bucketJson.toString());
                                            if (bucketJson.containsKey(NONMSGSTMCLSFLAG)
                                                    && bucketJson.getIntValue(NONMSGSTMCLSFLAG) == 1) {
                                                //符合非信息流修正先置为0
                                                bucketJson.put(CORRECTFLAG, 0);
                                                String sql = null;
                                                JSONArray channelArray = bucketJson.getJSONArray(CHANNELPRITORY);
                                                int cmsChannelArraySize = channelArray.size();
                                                if (cmsChannelArraySize > 0) {
                                                    int action = bucketJson.getIntValue(ACTIONS);
                                                    int adChose = bucketJson.getIntValue(ADCHSRULE);
                                                    LOG.info(
                                                            "deviceId: "
                                                                    + deviceId
                                                                    + " action: "
                                                                    + action
                                                                    + " adChose: "
                                                                    + adChose);
                                                    // 筛选点击还是曝光
                                                    if (action == 1) {
                                                        // 筛选最早还是最晚
                                                        if (adChose == 0) {
                                                            LOG.info("sql exp asc");
                                                            sql = SqlTemplate.expSqlTmplateAsc;
                                                        } else {
                                                            LOG.info("sql exp desc");
                                                            sql = SqlTemplate.expSqlTmplateDesc;
                                                        }
                                                    } else if (action == 2) {
                                                        if (adChose == 0) {
                                                            LOG.info("sql clk asc");
                                                            sql = SqlTemplate.clkSqlTmplateAsc;
                                                        } else {
                                                            LOG.info("sql clk desc");
                                                            sql = SqlTemplate.clkSqlTmplateDesc;
                                                        }
                                                    }

                                                    connection = DruidUtils.getConnection();
                                                    statement = connection.prepareStatement(sql);

                                                    statement.setString(1, deviceId);
                                                    statement.setInt(2, bucketJson.getInteger(TIMEBEFORE));
                                                    statement.setInt(3, adChose);

                                                    resultSet = statement.executeQuery();

                                                    HashMap<String, String> queryMap = new HashMap<>();

                                                    while (true) {
                                                        boolean sqlRes = resultSet.next();
                                                        LOG.info(
                                                                "##sqlRes: "
                                                                        + sqlRes
                                                                        + " deviceId: "
                                                                        + deviceId
                                                                        + " uid: "
                                                                        + uid);
                                                        if (!sqlRes) {
                                                            break;
                                                        }
                                                        // 多渠道优先级处理need add
                                                        String channel = resultSet.getString(CHANNEL);
                                                        String hfqOrigin = resultSet.getString(HFQORIGIN);
                                                        String hfqClkId = resultSet.getString(HFQCLKID);
                                                        LOG.info(
                                                                "sql Res deviceId:"
                                                                        + deviceId
                                                                        + " uid: "
                                                                        + uid
                                                                        + " channel: "
                                                                        + channel
                                                                        + " hfqOrigin:"
                                                                        + hfqOrigin);
                                                        LOG.info("sql Res: hfqClkId: " + hfqClkId);
                                                        queryMap.put(
                                                                channel, hfqOrigin + SPLITKEY + hfqClkId);
                                                    }
                                                    LOG.info("deviceId: " + deviceId + " uid: " + uid
                                                            + " queryMap size: "
                                                            + queryMap.size());

                                                    if (queryMap.size() > 0) {
                                                        for (int ix = 0; ix < cmsChannelArraySize; ix++) {
                                                            String channel = channelArray.getString(ix);
                                                            LOG.info(
                                                                    "cmsChannel: "
                                                                            + channel
                                                                            + " queryMap contains channel: "
                                                                            + queryMap.containsKey(
                                                                            channel));
                                                            if (queryMap.containsKey(channel)) {
                                                                // Channel其实并不需要，先加上吧
                                                                String mapValue = queryMap.get(channel);
                                                                LOG.info(
                                                                        "MapValue:"
                                                                                + (mapValue == null
                                                                                ? "null"
                                                                                : mapValue));
                                                                if (mapValue != null) {
                                                                    String[] splits = mapValue.split(SPLITKEY);
                                                                    LOG.info("splits.length: " + splits.length);
                                                                    if (splits.length == 2) {
                                                                        bucketJson.put(CHANNEL, channel);
                                                                        LOG.info("split[0]: " + splits[0]);
                                                                        if (splits[0] != null && !("null").equals(splits[0])) {
                                                                            bucketJson.put(CORRECTORIGINSRC, splits[0]);
                                                                            bucketJson.put(CORRECTFLAG, 1);
                                                                        }
                                                                        LOG.info("split[1]: " + splits[1]);
                                                                        bucketJson.put(CLICKID, splits[1]);
                                                                        bucketJson.put(CORRECTENDTIME,
                                                                                System.currentTimeMillis());
                                                                    }
                                                                }
                                                                break;
                                                            }
                                                        }
                                                    }
                                                }
                                            }

                                        }
                                    }
                                } catch (Exception e) {
                                    LOG.error("exception: " + e.toString());
                                } finally {
                                    if (resultSet != null) {
                                        resultSet.close();
                                    }

                                    if (statement != null) {
                                        statement.close();
                                    }
                                    LOG.info("sink json: " + resJson.toString());
                                    return resJson;
                                }
                            }
                        });

        CompletableFuture.supplyAsync(
                        new Supplier<JSONObject>() {
                            @Override
                            public JSONObject get() {
                                try {
                                    return droisResult.get();
                                } catch (Exception e) {
                                    LOG.error("supplyAsync exception: " + e.toString());
                                    return null;
                                }
                            }
                        })
                .thenAccept(
                        (JSONObject json) -> {
                            resultFuture.complete(Collections.singleton(json));
                        });
    }

    private JSONObject correctAttributionProcess(UserInfo user) {
        JSONObject resJson = new JSONObject();
        try {
            LOG.info("##ruleInfo: " + ruleInfo.toString());
            long actionTime = user.getActionTime();
            String userKey = user.getUserKey();
            long uid = user.getUid();
            String deviceId = user.getDeviceId();
            String userAction = user.getAction();
            resJson.put(USERKEY, userKey);
            resJson.put(UID, uid);
            resJson.put(DEVICEID, deviceId);
            resJson.put(BIZID, user.getBizId());
            resJson.put(ACTION, userAction); //node or audit
            resJson.put(ACTIONTIME, user.getActionTime());
            JSONArray attributionRes = new JSONArray();

            long requesTime = ruleInfo.getRequesTime();
            Integer dateDiff = CommonTimeTools.dateDiff(actionTime * 1000, requesTime, sdf);
            LOG.info("dateDiff: " + dateDiff + " requesTime: "
                    + requesTime + " actionTime: " + actionTime);

            if (dateDiff > 0) {
                // 更新CMS配置
                CmsRuleInfo tmpCmsRule = getCmsRuleInfo();
                //如果新获取的cmsRule有问题，沿用之前的cmsRule配置
                if(tmpCmsRule != null) {
                    ruleInfo = tmpCmsRule;
                }
            }
            LOG.info("Use ruleInfo: " + ruleInfo.toString());
            LOG.info("UserAction: " + userAction);
            LinkedList<BucketInfo> bucketInfoLinkedList = ruleInfo.getAttribuStgMap().get(userAction);
            if(ruleInfo.isNodeEnabel(userAction) && bucketInfoLinkedList != null) {
                int listSize = bucketInfoLinkedList.size();
                BucketInfo sampleBucketInfo = bucketInfoLinkedList.getFirst();
                resJson.put(NODE, sampleBucketInfo.getNode());
                resJson.put(BUCKETSTRATEGYID, sampleBucketInfo.getBktStrategyId());
                resJson.put(BUCKETSTRATEGYNAME, sampleBucketInfo.getBktStrategyName());
                int mode = sampleBucketInfo.getMode();
                resJson.put(MODE, mode);
                if (mode == 1) {
                    //userBktNum 取值范围0-90
                    int userBktNum = (user.getCurCnt() % 10) * 10;
                    int percentSum = 0;
                    for (int i = 0; i < listSize; i++) {
                        BucketInfo bucketInfo = bucketInfoLinkedList.get(i);
                        int percent = bucketInfo.getPercent();
                        //找到对应的桶
                        if (userBktNum >= percentSum && userBktNum < (percentSum + percent)) {
                            LOG.info("userKey: " + userKey + " uid: " + uid + " deviceId: " + deviceId +
                                    " find bucket id: " + bucketInfo.getBucketId() +
                                    " bucket name: " +bucketInfo.getBucketName());
                            attributionRes.add(constructAttribuRes(user, bucketInfo));
                            break;
                        }
                        percentSum += percent;
                    }
                } else if(mode == 2) {
                    for (int i = 0; i < listSize; i++) {
                        BucketInfo bucketInfo = bucketInfoLinkedList.get(i);
                        attributionRes.add(constructAttribuRes(user, bucketInfo));
                    }
                } else {
                    LOG.error("getAttribuStgMap mode error : " + mode);
                }
            } else {
                LOG.error("getAttribuStgMap does not contains list for userAction: " + userAction);
            }
            resJson.put(ATTRBURES, attributionRes);
        } catch (Exception e) {
            LOG.error("correctAttributionProcess Excep: " + e.toString());
        }
        return resJson;
    }

    private JSONObject constructAttribuRes(UserInfo user, BucketInfo bucketInfo) {
        JSONObject resJson = new JSONObject();
        resJson.put(ATRBUCKETID, bucketInfo.getBucketId());
        resJson.put(ATRBUCKETNAME, bucketInfo.getBucketName());
        resJson.put(POSTBACKID, bucketInfo.getPostBackId());
        resJson.put(ATRIBUSTTID, bucketInfo.getAttribuId());
        resJson.put(ATRIBUSTTNAME, bucketInfo.getAttribuName());

        //add
        resJson.put(CHANNELPRITORY, bucketInfo.getChannel());
        resJson.put(ADCHSRULE, bucketInfo.getAdChooseRule());
        resJson.put(ACTIONS, bucketInfo.getSupportActions());
        resJson.put(TIMEBEFORE, bucketInfo.getDateDiff());

        resJson.put(CORRECTFLAG, 2);
        long curtime = System.currentTimeMillis();
        resJson.put(CORRECTSTIME, curtime);
        resJson.put(CORRECTENDTIME, curtime);
        resJson.put(NONMSGSTMCLSFLAG, 2);
        resJson.put(ABTESTFLAG, 2);

        // 只取前5位和CMS匹配
        String originSource = user.getOriginSource();
        String type = user.getType();
        int nonMsgStreamClassFlag = 0;
        if (type != null) {
            if ("MA".equals(type)
                    && bucketInfo.getMa() != null
                    && ("__all__".equals(bucketInfo.getMa()))) {
                nonMsgStreamClassFlag = 1;
            } else if ("H5".equals(type) && bucketInfo.getH5() != null) {
                if (originSource != null) {
                    // 取前5位进行匹配
                    String subOriginSource =
                            (originSource.length() >= 5)
                                    ? originSource.substring(0, 5)
                                    : originSource;
                    String h5 = bucketInfo.getH5();
                    // CMS H5没有配置是''空
                    if (h5 != null && !"".equals(h5) && h5.contains(subOriginSource)) {
                        nonMsgStreamClassFlag = 1;
                    }
                }
            }
            resJson.put(NONMSGSTMCLSFLAG, nonMsgStreamClassFlag);

            // 0 对照组 1 实验组 2 null
            int abTestFlag = 2;
            if (nonMsgStreamClassFlag == 1) {
                long uidTail = user.getUid() % 10;
                String uids = bucketInfo.getUids();
                if (uids != null && uids.contains(uidTail + "")) {
                    abTestFlag = 1;
                } else {
                    abTestFlag = 0;
                }
            }
            resJson.put(ABTESTFLAG, abTestFlag);
        }

        return resJson;
    }

    @Override
    public void timeout(UserInfo user, ResultFuture<JSONObject> resultFuture) throws Exception {
        LOG.error(
                "DorisAsyncRichMapFunction time out: "
                        + " deviceId: "
                        + user.getDeviceId()
                        + " userKey: "
                        + user.getUserKey());

        // need to modify collect
        JSONObject timeOut = new JSONObject();
        timeOut.put(CORRECTFLAG, 2);
        timeOut.put(USERKEY, user.getUserKey());
        timeOut.put(UID, user.getUid());
        timeOut.put(DEVICEID, user.getDeviceId());
        timeOut.put("error", "timeout to get attribute result!!");
        resultFuture.complete(Collections.singleton(timeOut));
    }
}
